from pyspark.sql import SparkSession
import os
import pyspark.sql.functions as F
from cn.itcast.tag.base.BaseModel import BaseModel



# 0.设置系统环境变量
os.environ['JAVA_HOME'] = '/export/server/jdk1.8.0_241/'
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/envs/pyspark_env/bin/python'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/envs/pyspark_env/bin/python'

class ConsumerCycleModel(BaseModel):
    # inType = Elasticsearch  ##esNodes=up01:9200##esIndex=tfec_tbl_orders##esType=_doc##selectFields=memberid,finishtime
    def compute(self, es_df, five_df):
        #todo:1-先统计每个用户最近的一次订单时间
        es_df = es_df.groupBy("memberid").agg(F.max("finishtime").alias("finishtime"))
        #todo:2-将时间戳转换为日期，并计算与当前时间的差值
        #转换为年-月-日的方式后，用于计算时间天数差值
        #current_date()：当前时间（年-月-日）
        #from_unixtime()：把timestamp时间戳类型的数据转换为指定的时间格式
        #datediff()：时间差值函数
        #date_sub()：时间相减的函数
        #date_add()：时间相加的函数
        #date_format()：时间格式转换

        """
        
          memeberid   current_date    finishtime     days
            1            2025-04-07      2021-04-04   1207
        
        """
        es_df = es_df.select("memberid",
                             F.current_date().alias("current_date"),
                             F.from_unixtime("finishtime", format='yyyy-MM-dd').alias("finishtime"),
                             F.datediff(F.current_date(),F.from_unixtime("finishtime",format='yyyy-MM-dd')).alias("days"))
        es_df = es_df.select("memberid","days")
        es_df.show()

        #todo:3-将标签的时间周期构建两列时间范围，start、end
        #split(str,pattern)：切割函数
        five_df = five_df.select("id",
                       F.split("rule","-")[0].alias("start"),
                       F.split("rule","-")[1].alias("end"))
        five_df.show()
        #todo:4-根据用户订单时间差匹配标签范围获取标签
        #join中带有between的写法(start,end)
        new_df = es_df.join(five_df,es_df['days'].between(five_df['start'],five_df['end']),'left')\
            .select(es_df['memberid'].alias("userId"),five_df['id'].alias("tagsId"))
        new_df.show()
        return new_df

if __name__ == '__main__':
    model = ConsumerCycleModel(23)
    model.execute()